package com.ber.rocketmq;

import com.ber.rocketmq.domain.Msg;
import com.ber.rocketmq.service.MsgService;
import com.ber.rocketmq.service.impl.MsgServiceImpl;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Properties;

/**
 * @Author bq8023
 * @Date 21/7/6 10:08
 * @Version 1.0
 */
@Component
public class Consumer {
    @Autowired
    private MsgService msgService;

    @PostConstruct
    public void consumer() {
        String topic = "test";
        String tag = "a";
        String group = "test_group";
        try {
            /*
             * Instantiate with specified consumer group name.
             */
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);

            // 设置NameServer地址
            consumer.setNamesrvAddr("58.59.43.108:9876");
            // 订阅Topic
            //设置consumer所订阅的Topic和Tag，*代表全部的Tag
            consumer.subscribe(topic, tag);
            //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，跳过历史消息
            //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            /*
             *  回调接口，处理获取到的消息
             * //MessageListenerOrderly 这个是有序的
            //MessageListenerConcurrently 这个是无序的,并行的方式处理，效率高很多
             */
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, contect) -> {
                System.err.printf("%s Receive New Messages: %n", Thread.currentThread().getName());
                try {
                    for (MessageExt msg : msgs) {
                        System.out.println("Keys-------->" + msg.getKeys());
                        System.out.println("MsgId-------->" + msg.getMsgId());
                        System.out.println("QueueId-------->" + msg.getQueueId());
                        System.out.println("QueueOffset-------->" + msg.getQueueOffset());
                        System.out.println("Body-------->" + new String(msg.getBody()));
                        Msg savemsg = new Msg();
                        savemsg.setBody(new String(msg.getBody()));
                        savemsg.setMsgId(msg.getMsgId());
                        savemsg.setQueueId((long) msg.getQueueId());
                        savemsg.setQueueOffset(msg.getQueueOffset());
                        msgService.saveMsg(savemsg);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    //稍后再试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });

            /*
             *  Launch the consumer instance.
             */
            consumer.start();
            System.out.printf("Consumer Started.%n");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void main(String[] args) throws Exception {


        String topic = "test";
        String tag = "a";
        String group = "test_group";

        /*
         * Instantiate with specified consumer group name.
         */
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
        // 设置NameServer地址
        consumer.setNamesrvAddr("58.59.43.108:9876");
        // 订阅Topic
        consumer.subscribe(topic, tag);
        // CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，跳过历史消息
        // CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /*
         *  回调接口，处理获取到的消息
         * MessageListenerOrderly 这个是有序的
         * MessageListenerConcurrently 这个是无序的,并行的方式处理，效率高很多
         */
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                System.err.printf("%s Receive New Messages: %n", Thread.currentThread().getName());
                for (MessageExt msg : msgs) {
                    System.out.println("Keys-------->" + msg.getKeys());
                    System.out.println("MsgId-------->" + msg.getMsgId());
                    System.out.println("QueueId-------->" + msg.getQueueId());
                    System.out.println("QueueOffset-------->" + msg.getQueueOffset());
                    System.out.println("Body-------->" + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        /*
         *  Launch the consumer instance.
         */
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}
